package drds.global_transaction.api.rpc.rpc_server;

import drds.global_transaction.api.api.RpcContext;
import drds.global_transaction.api.in_bound_handler.TransactionMessageHandler;
import drds.global_transaction.api.message_codec.HeartbeatMessage;
import drds.global_transaction.api.message_codec.request_and_response.register_manager.resource_manager.RegisterResourceManagerRequestMessage;
import drds.global_transaction.api.message_codec.request_and_response.register_manager.transaction_manager.RegisterTransactionManagerRequestMessage;
import drds.global_transaction.api.rpc.config.ServerConfig;
import drds.global_transaction.api.rpc.rpc_message.RpcInfo;
import drds.global_transaction.common.exception.FrameworkErrorCode;
import drds.global_transaction.common.exception.FrameworkException;
import drds.global_transaction.common.util.Nets;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandler.Sharable;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeoutException;

@Slf4j
@Sharable
public class RpcServer extends AbstractRpcServer implements MessageSend
{
	@Setter
	@Getter
	protected TransactionMessageListener transactionMessageListener;
	@Setter
	@Getter
	private TransactionMessageHandler transactionMessageHandler;
	@Setter
	@Getter
	private RegisterCheckAuthHandler registerCheckAuthHandler;

	public RpcServer(ThreadPoolExecutor messageExecutor)
	{
		super(new ServerConfig(), messageExecutor);
	}

	public void setHandler(TransactionMessageHandler transactionMessageHandler)
	{
		setHandler(transactionMessageHandler, null);
	}

	public void setHandler(TransactionMessageHandler transactionMessageHandler, RegisterCheckAuthHandler registerCheckAuthHandler)
	{
		this.transactionMessageHandler = transactionMessageHandler;
		this.registerCheckAuthHandler = registerCheckAuthHandler;
	}

	//
	@Override
	public void init()
	{
		super.init();
		setChannelHandlers(new ChannelHandler[] { RpcServer.this });
		TransactionMessageListenerImpl messageListener = new TransactionMessageListenerImpl(transactionMessageHandler);
		messageListener.setMessageSend(this);
		this.setTransactionMessageListener(messageListener);
		super.start();

	}

	@Override
	public void destroy()
	{
		super.destroy();
		super.shutdown();
		if (log.isInfoEnabled())
		{
			log.info("destroyed rpcServer");
		}
	}

	//

	@Override
	public void channelRead(final ChannelHandlerContext channelHandlerContext, Object message) throws Exception
	{
		if (message instanceof RpcInfo)
		{
			RpcInfo rpcInfo = (RpcInfo) message;
			// 为什么是在channelRead进行RegisterTransactionManagerRequestMessage
			if (rpcInfo.getMessage() instanceof RegisterTransactionManagerRequestMessage)
			{
				RegisterTransactionManagerRequestMessage registerTransactionManagerRequestMessage = (RegisterTransactionManagerRequestMessage) rpcInfo.getMessage();
				transactionMessageListener.onRegisterTransactionManagerRequest(channelHandlerContext, this, rpcInfo.getId(), registerTransactionManagerRequestMessage, registerCheckAuthHandler);
				return;
			}
			if (rpcInfo.getMessage() == HeartbeatMessage.ping)
			{
				transactionMessageListener.onPing(channelHandlerContext, this, rpcInfo.getId());
				return;
			}
		}
		super.channelRead(channelHandlerContext, message);
	}

	@Override
	public void dispatch(ChannelHandlerContext channelHandlerContext, long messageId, Object message)
	{
		if (message instanceof RegisterResourceManagerRequestMessage)
		{
			transactionMessageListener.onRegisterResourceManagerRequest(channelHandlerContext, this, messageId, (RegisterResourceManagerRequestMessage) message, registerCheckAuthHandler);
			// 进行交互的前提是注册成功
		} else
		{
			if (ChannelManager.isRegistered(channelHandlerContext.channel()))
			{
				transactionMessageListener.onMessage(channelHandlerContext, this, messageId, message);
			} else
			{
				try
				{
					closeChannelHandlerContext(channelHandlerContext);
				} catch (Exception e)
				{
					log.error(e.getMessage());
				}
				if (log.isInfoEnabled())
				{
					log.info(String.format("close a unhandled connection! [%s]", channelHandlerContext.channel().toString()));
				}
			}
		}
	}
	//

	private void closeChannelHandlerContext(ChannelHandlerContext channelHandlerContext)
	{
		if (log.isInfoEnabled())
		{
			log.info("closeChannelHandlerContext channel:" + channelHandlerContext.channel());
		}
		channelHandlerContext.disconnect();
		channelHandlerContext.close();
	}

	//
	@Override
	public void userEventTriggered(ChannelHandlerContext channelHandlerContext, Object event)
	{
		if (event instanceof IdleStateEvent)
		{
			IdleStateEvent idleStateEvent = (IdleStateEvent) event;
			if (idleStateEvent.state() == IdleState.READER_IDLE)
			{
				rpcContextRelease(channelHandlerContext);
				try
				{
					closeChannelHandlerContext(channelHandlerContext);
				} catch (Exception e)
				{
					log.error(e.getMessage());
				}
			}
		}
	}

	@Override
	public void channelInactive(ChannelHandlerContext channelHandlerContext) throws Exception
	{
		if (threadPoolExecutor.isShutdown())
		{
			return;
		}
		rpcContextRelease(channelHandlerContext);
		super.channelInactive(channelHandlerContext);
	}

	private void rpcContextRelease(ChannelHandlerContext channelHandlerContext)
	{
		final String ipAndPort = Nets.getHostAddressAndPort(channelHandlerContext.channel().remoteAddress());
		RpcContext rpcContext = ChannelManager.getRpcContext(channelHandlerContext.channel());
		if (log.isInfoEnabled())
		{
			log.info(ipAndPort + " to transaction_coordinator channel inactive.");
		}
		if (null != rpcContext && null != rpcContext.getRole())
		{
			rpcContext.release();// rpcContext release
			if (log.isInfoEnabled())
			{
				log.info("remove channel:" + channelHandlerContext.channel() + "global_transaction_id_context:" + rpcContext);
			}
		} else
		{
			log.warn("remove unused channel:" + channelHandlerContext.channel());
		}
	}

	//
	@Override
	public void exceptionCaught(ChannelHandlerContext channelHandlerContext, Throwable cause) throws Exception
	{
		if (log.isInfoEnabled())
		{
			log.info("channel exx:" + cause.getMessage() + ",channel:" + channelHandlerContext.channel());
		}
		ChannelManager.releaseRpcContext(channelHandlerContext.channel());
		super.exceptionCaught(channelHandlerContext, cause);
	}

	//
	@Override
	public void synchronousSendRequest(String clientIp, String applicationId, String resourceId, Object request)
	{
		Channel channel = ChannelManager.getChannel(resourceId, applicationId, clientIp);
		if (channel == null)
		{
			throw new FrameworkException("resource_manager_client client is not connected. dbkey:" + resourceId + ",clientIp:" + clientIp);
		} else
		{
			try
			{
				super.synchronousSendRequest(channel, request);
			} catch (FrameworkException e)
			{
				if (e.getFrameworkErrorCode() == FrameworkErrorCode.ChannelIsNotWritable)
				{
					ChannelManager.releaseRpcContext(channel);
					throw e;
				}
			}
		}
	}

	@Override
	public Object asynchronousSendRequestWithTimeoutGetResponse(String clientIp, String applicationId, String resourceId, Object message) throws TimeoutException
	{
		return asynchronousSendRequestWithTimeoutGetResponse(clientIp, applicationId, resourceId, message, ServerConfig.getRpcRequestTimeout());
	}

	@Override
	public Object asynchronousSendRequestWithTimeoutGetResponse(String clientIp, String applicationId, String resourceId, Object message, long timeout) throws TimeoutException
	{
		Channel channel = ChannelManager.getChannel(resourceId, applicationId, clientIp);
		if (channel != null)
		{
			return super.asynchronousSendRequestWithTimeoutGetResponse(null, channel, message, timeout);
		} else
		{
			throw new RuntimeException("resource_manager_client client is not connected. dbkey:" + resourceId + ",clientIp:" + clientIp);
		}
	}

	//
	@Override
	public void synchronousSendResponse(Channel channel, long messageId, Object message)
	{
		Channel $channel = channel;
		if (!(message instanceof HeartbeatMessage))
		{
			$channel = ChannelManager.getChannel(channel);
		}
		if ($channel != null)
		{
			super.synchronousSendResponse($channel, messageId, message);
		} else
		{
			throw new RuntimeException("channel is error. channel:" + $channel);
		}
	}

	/**
	 * Send response. redress,merge msg
	 */
	@Override
	public void synchronousSendResponse(String clientIp, String applicationId, String resourceId, long messageId, Object message)
	{
		Channel channel = ChannelManager.getChannel(resourceId, applicationId, clientIp);
		if (channel != null)
		{
			try
			{
				super.synchronousSendResponse(channel, messageId, message);
			} catch (FrameworkException e)
			{
				if (e.getFrameworkErrorCode() == FrameworkErrorCode.ChannelIsNotWritable)
				{
					log.error("channel is not writeable,channel:" + channel);
					ChannelManager.releaseRpcContext(channel);
				}
				throw e;
			}
		} else
		{
			throw new RuntimeException("channel is error. channel:" + channel);
		}
	}

}
